/*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.analysys.presto.connector.hbase.adaptive;

import com.analysys.presto.connector.hbase.meta.HBaseColumnMetadata;
import com.analysys.presto.connector.hbase.meta.HBaseConfig;
import com.facebook.airlift.log.Logger;
import com.facebook.presto.spi.ColumnMetadata;
import com.facebook.presto.spi.adaptive.metadata.AdaptiveConnectorMetadaManager;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.List;

import static com.analysys.presto.connector.hbase.utils.Utils.matchType;

public class HMetadataManager
        implements AdaptiveConnectorMetadaManager
{
    private static final Logger log = Logger.get(HMetadataManager.class);

    private String ZK_HOST;
    private String HBASE_METADATA_PATH;
    private String isAdaptiveEnable = "true";

    private CuratorFramework client = null;
    private HMetadata hMetadata;

    public HMetadataManager(HBaseConfig hbaseConfig)
    {
        this.ZK_HOST = hbaseConfig.getMetadataZKHost();
        this.HBASE_METADATA_PATH = hbaseConfig.getMetadataZKNode();
        this.isAdaptiveEnable = hbaseConfig.getIsAdaptiveEnable();

        client = CuratorFrameworkFactory
                .builder()
                .connectString(ZK_HOST)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .build();
        client.start();

        this.hMetadata = readHBaseMetadataFromZK();
    }

    public boolean isAdaptiveEnable()
    {
        return isAdaptiveEnable.equals("true");
    }

    /**
     * Add a table to HBase and write metadata to zk.
     */
    public void addATable(HTableInfo hTableInfo)
    {
        hMetadata.gethTableInfoList().add(hTableInfo);
        writeHBaseMetadataToZK(hMetadata);
    }

    /**
     * remove a table from HBase and write metadata to zk.
     *
     * @param schemaName schemaName
     * @param tableName  tableName
     */
    public void removeATable(String schemaName, String tableName)
    {

        hMetadata.gethTableInfoList().remove(new HTableInfo(schemaName, tableName));
        writeHBaseMetadataToZK(hMetadata);
    }

    /**
     * remove a table from HBase and write metadata to zk.
     *
     * @param tableName tableName
     */
    public void removeATable(String tableName)
    {

        hMetadata.gethTableInfoList().remove(new HTableInfo(tableName));
        writeHBaseMetadataToZK(hMetadata);
    }

    /**
     * Read data and deserialize HBaseMetadata from zk.
     * If zkNode is null, it will return a hbaseMetadata including empty HTable list.
     */
    private HMetadata readHBaseMetadataFromZK()
    {
        try {
            return HMetadataUtil.deserialize(client.getData().forPath(HBASE_METADATA_PATH));
        }
        catch (Exception e) {
            log.error("HBaseMetadata read from ZK failed, " + e.getMessage());
            return new HMetadata();
        }
    }

    /**
     * Serialize HBaseMetadata and write data to zk.
     * If zkNode is null, it will create a node and write it.
     */
    private void writeHBaseMetadataToZK(HMetadata hMetadata)
    {
        try {

            if (client.checkExists().forPath(HBASE_METADATA_PATH) == null) {
                client.create().creatingParentsIfNeeded().forPath(HBASE_METADATA_PATH, HMetadataUtil.serialize(hMetadata));
            }
            else {
                client.setData().forPath(HBASE_METADATA_PATH, HMetadataUtil.serialize(hMetadata));
            }
        }
        catch (Exception e) {
            log.error("HBaseMetadata write to ZK failed, " + e.getMessage());
        }
    }

    /**
     * Get table list.
     */
    public List<HTableInfo> getTableList()
    {
        return hMetadata.gethTableInfoList();
    }

    @VisibleForTesting
    public HMetadataManager(String zkHost, String zkNode)
    {
        this.ZK_HOST = zkHost;
        this.HBASE_METADATA_PATH = zkNode;

        client = CuratorFrameworkFactory
                .builder()
                .connectString(ZK_HOST)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .build();
        client.start();

        this.hMetadata = readHBaseMetadataFromZK();
    }

    /**
     * Get table by schema.table
     */
    public HTableInfo getTable(String schemaName, String tableName)
    {
        return
                this.getTableList().stream()
                        .filter(table -> table.equals(schemaName, tableName))
                        .findFirst()
                        .orElse(null);
    }

    /**
     * Get column metadata by schema.table
     */
    public ImmutableList<ColumnMetadata> gethColumnMetaList(String schemaName, String tableName)
    {
        ImmutableList.Builder<ColumnMetadata> columnsMetadata = ImmutableList.builder();
        try {
            HTableInfo hTableInfo = this.getTable(schemaName, tableName);
            List<HColumn> hColumns = hTableInfo.gethColumnList();
            boolean hasRowKey = false;
            for (HColumn hColumn : hColumns) {
                String family = hColumn.getFamily();
                String columnName = hColumn.getColumnName();

                String type = hColumn.getType();
                boolean isRowKey = hColumn.isRowKey();
                columnsMetadata.add(new HBaseColumnMetadata(family, columnName, matchType(type), isRowKey));
                if (isRowKey) {
                    hasRowKey = true;
                }
            }
            if (!this.isAdaptiveEnable()) {
                Preconditions.checkState(hasRowKey, "Table %s.%s doesn't specified ROW_KEY column. Specify ROW_KEY in %s,%s.",
                        schemaName,
                        tableName,
                        ZK_HOST,
                        HBASE_METADATA_PATH);
            }
        }
        catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        return columnsMetadata.build();
    }
}